package api

import (
	"gitee.com/chejiangyi/bsfgo/core/utils"
	"gitee.com/chejiangyi/bsfgo/gin2/controller"
	"gitee.com/chejiangyi/bsfgo/kafka2"
	"github.com/gin-gonic/gin"
	"time"
)

type KafkaController struct {
	controller.BaseApiController
}

func (m *KafkaController) Producer(c *gin.Context) {
	defer m.CatchJsonError(c)
	producer := kafka2.KafkaUtil.CreateProducer()
	defer producer.Close()
	for i := 0; i < 1; i++ {
		producer.SendMessage("test", "test_key"+time.Now().String(), utils.JsonUtil.ToJson(map[string]any{"aaa": "bbb" + time.Now().String()}))
	}
	m.JsonSuccess(c, "ok")
}

var consumer *kafka2.KafkaConsumer

func (m *KafkaController) Consumer(c *gin.Context) {
	defer m.CatchJsonError(c)
	if consumer == nil {
		consumer := kafka2.KafkaUtil.CreateConsumerGroup()
		//defer consumer.Close()
		consumer.Consume([]string{"test"}, func(message *kafka2.KafkaMessage, event string) {
			defer func() {
				if err := recover(); err != nil {
					utils.LogUtil.Error(err.(error), "消息处理报错:key %v,event %v", message.Key, event)
					//单条消息报错,默认处理后不抛错，消息继续消费（可丢失消息）
					/*do nothing*/
					//单条消息报错，处理后继续抛错，消息终止消费（重要消息）
					panic(err)
				}
			}()
			if event == "setup" {
				//启动
				//此处可以手工重置offset
				//message.Session.ResetOffset() 重置offset
			}
			if event == "cleanup" {
				//关闭
				//此处可以回收关闭资源
			}
			if event == "message" {
				utils.LogUtil.Info("topic:%v,partition:%v,message:%v,offset:%v,key:%v",
					message.Topic, message.Partition, string(message.Value), message.Offset, string(message.Key))
			}
		})
	}
	m.JsonSuccess(c, "ok")
}
